package com.navinfo.opentsp.platform.computing.serializer;

import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCAMTDataUpload;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;

/**
 * amtDataUpload信息
 * HDFS_Sink的amtDataUpload序列化器，拼接成字段间通过\001分割的字符串
 */
public class HDFSAMTDataUploadSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSAMTDataUploadSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private HDFSAMTDataUploadSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    @Override
    public boolean supportsReopen() {
        return false;
    }

    @Override
    public void afterCreate() {
    }

    @Override
    public void afterReopen() {
    }

    @Override
    public void beforeClose() {
    }

    @Override
    public void flush() throws IOException {
        this.out.flush();
    }

    @Override
    public void write(Event event) throws IOException {
        try {
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            LCAMTDataUpload.AMTDataUpload amtDataUpload = LCAMTDataUpload.AMTDataUpload.parseFrom(event.getBody());
            StringBuilder amtDataUploadStr = new StringBuilder(5000);
            amtDataUploadStr.append(amtDataUpload.getLatitude()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getLongitude()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getOriginalLat()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getOriginalLng()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getHeight()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getDirection()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGpsDate()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getAccelerator()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getDriverRequestEngineMode()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getRotation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getEngineOutputTorque()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTransmissionOutputShaftRotation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getClutchSlipRatio()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTransmissionInputShaftRotation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTransmissionSystemCombinedState()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGear()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGearboxMode1()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGearboxMode2()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGearboxMode3()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGearboxMode4()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getGearboxRequestGear()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getSpeed()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTotalMileage()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getExhaustBrakingState()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getFuelConsumption()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getInstantaneousFuelConsumption()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getAverageFuelConsumption()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTemporaryManualRequest()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getBrake()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getCruiseActivation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getParkingBrake()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getLeftFrontWheelRotation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getRightFrontWheelRotation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getABSActivation()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getRampStartSwitch()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getAtmosphericPressure1()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getAtmosphericTemperature()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getAirPressure()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getQuality()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getRamp()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTransmissionOilTemperature()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getClutchOffset()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getClutchWearPercentage()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getClutchTemperature()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getClutchNipPoint()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getTerminalId()).append(splitStr);
            amtDataUploadStr.append(amtDataUpload.getSerialNumber()).append(splitStr);

            //添加字段只能在分割线之上添加,注意时间分区这块，终端传的时候是秒还是毫秒
            //------------------------------------------分割线----------------------------------------
            amtDataUploadStr.append(DateUtils.format(amtDataUpload.getGpsDate() * 1000, "yyyyMMdd"));

            //将数据拼接后写入hdfs
            this.out.write(amtDataUploadStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        } catch (Exception e) {
            logger.error("执行HDFS的HDFSAMTDataUploadSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

    public static class Builder implements EventSerializer.Builder {
        public Builder() {
        }

        @Override
        public EventSerializer build(Context context, OutputStream out) {
            return new HDFSAMTDataUploadSerializer(out, context);
        }
    }
}
